package com.guanzi.netty.client;

import com.guanzi.netty.message.RpcRequestMessage;
import com.guanzi.netty.protocol.MessageCodecSharable;
import com.guanzi.netty.protocol.ProcotolFrameDecoder;
import com.guanzi.netty.protocol.SequenceIdGenerator;
import com.guanzi.netty.server.handler.RpcResponseMessageHandler;
import com.guanzi.netty.server.service.HelloService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.SequenceInputStream;
import java.lang.reflect.Proxy;
import java.util.concurrent.TimeUnit;


public class RpcClientManager {

    private static final Logger log = LoggerFactory.getLogger(RpcClientManager.class);
    private static volatile Channel channel = null;
    private static final Object LOCK = new Object();

    public static void main(String[] args) {
        HelloService service = getProxyService(HelloService.class);
        String msg = service.sayHello("guanzi_123");
        System.out.println(msg);
    }

    // 双重校验单例锁 获取Channel
    public static Channel getChannel() {
        if (channel != null) {
            return channel;
        }
        synchronized (LOCK) {
            if (channel != null) {
                return channel;
            }
            initChannel();
            return channel;
        }
    }

    /**
     * 初始化channel
     */
    private static void initChannel() {
        NioEventLoopGroup group = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
        RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.group(group);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ProcotolFrameDecoder());
                ch.pipeline().addLast(LOGGING_HANDLER);
                ch.pipeline().addLast(MESSAGE_CODEC);
                ch.pipeline().addLast(RPC_HANDLER);
            }
        });

        try {
            // 这里sync，是为了保证channel必须等到连接建立好才可以使用
            channel = bootstrap.connect("localhost", 8080).sync().channel();
            // 这里的addListener，是为了给channel添加以后的关闭逻辑，但不是要让它一直同步等着关闭，因为多个线程来了都还要使用channel进行主要的写
            channel.closeFuture().addListener(future -> {
                group.shutdownGracefully();
            });
        } catch (Exception e) {
            log.error("client error", e);
        }
    }

    // 创建代理类
    public static <T> T getProxyService(Class<?> serviceClass) {
        Class<?>[] interfaces = {serviceClass};
        // 这个代理类的作用：原始类的所有方法，经过代理类的方法调用时，都会经过下面这段自定义逻辑
        Object proxyInstance = Proxy.newProxyInstance(serviceClass.getClassLoader(), interfaces, (proxy, method, args) -> {
            // proxy 代理对象 method 代理方法 args所需参数

            int sequenceId = SequenceIdGenerator.nextId();
            // 1、将方法调用转换为消息对象
            RpcRequestMessage message = new RpcRequestMessage(
                    sequenceId,
                    serviceClass.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args);

            // 2、将消息对象发送出去
            getChannel().writeAndFlush(message);

            // 3、创建promise书包
            // 和写数据的nio线程，采用同一个eventLoop线程，这样下面就是该eventLoop线程拿着书包异步等结果了(promise.await())
            DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
            RpcResponseMessageHandler.PROMISES.put(sequenceId, promise);

            // 4、等待promise从RpcResponse带回来的结果
            // 4.1、RpcServer的RpcRequestHandler处理并写回RpcResponse数据
            // 4.2、RpcClientManager的channel入站，由RpcResponseHandler写入Promise结果
            // 4.3、await方法等回了promise数据，继续向下调用
            promise.await(30, TimeUnit.SECONDS);
            if (promise.isSuccess()) {
                // 调用正常
                return promise.getNow();
            } else {
                // 调用失败
                log.error("error", promise.cause());
                throw new RuntimeException(promise.cause());
            }
        });
        return (T) proxyInstance;
    }

}
